package kafka_model_apps

import java.sql.ResultSet

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

import scala.util.parsing.json.JSON

object MyKafka {


    def main(args: Array[String]) {
        println("环境搭建完成")
        Logger.getLogger("org").setLevel(Level.WARN)

        val conf = new SparkConf().setMaster("local[3]").setAppName("kafka")

        val ssc = new StreamingContext(conf, Seconds(5))
        ssc.checkpoint("D:\\项目\\streaming\\apps")

        val kafkaParams = Map[String, Object](
            "bootstrap.servers" -> "192.168.215.10:9092",
            "key.deserializer" -> classOf[StringDeserializer],
            "value.deserializer" -> classOf[StringDeserializer],
            "group.id" -> "kafka01"
        )

        val topics = Array("app_list")

        println("===========测试开始===========")
        var rs: ResultSet = null
        println("===========测试结束===========")
        val kafkaStream = KafkaUtils.createDirectStream[String, String](
            ssc,
            PreferConsistent,
            Subscribe[String, String](topics, kafkaParams))

        val ds: Unit = kafkaStream
            .map(app => (app.value()))
            .foreachRDD((rdd, index) => {
                println("---------------------------" + index)
                rdd.foreach(item => {
                    println(item)
                    val data = JSON.parseFull(item).get.asInstanceOf[Map[String, Any]]

                    val appData = new Apps(data)

                    println(appData.title)

                    val SELECT = s"SELECT id FROM apps WHERE title = '${appData.title}'"

                    println(s"SELECT:${SELECT}")

                    rs = myConn.conn().createStatement().executeQuery(SELECT)
                    if (!rs.next()) {
                        //数据库没有相同数据
                        println("not have")
                        val sql = s"INSERT INTO apps (" +
                            s"title," +
                            s"volume," +
                            s"downloads," +
                            s"follow," +
                            s"comment," +
                            s"tags," +
                            s"rank_num," +
                            s"rom," +
                            s"developer," +
                            s"rank_num_users," +
                            s"update_time)" +
                            s" VALUES (" +
                            s"'${appData.title}','${appData.volume}'," +
                            s"'${appData.downloads}','${appData.follow}'" +
                            s",'${appData.comment}','${appData.tags}'" +
                            s",'${appData.rank_num}','${appData.rom}'" +
                            s",'${appData.developer}','${appData.rank_num_users}'" +
                            s",'${appData.update_time}')"
                        println(s"insert:${sql}")
                        myConn.conn().createStatement().executeUpdate(sql)
                    } else {
                        //数据库有相同数据
                        println("has")
                        val sql = s"UPDATE apps" +
                            s" SET" +
                            s" volume = '${appData.volume}'," +
                            s" downloads = '${appData.downloads}'," +
                            s" follow = '${appData.follow}'," +
                            s" comment = '${appData.comment}'," +
                            s" tags = '${appData.tags}'," +
                            s" rank_num = '${appData.rank_num}'," +
                            s" rom = '${appData.rom}'," +
                            s" developer = '${appData.developer}'," +
                            s" rank_num_users = '${appData.rank_num_users}'," +
                            s" update_time = '${appData.update_time}'" +
                            s"WHERE title = '${appData.title}'"
                        println(s"updateSql:${sql}")
                        myConn.conn().createStatement().executeUpdate(sql)
                    }
                })
            })
        println(ds)

        ssc.start()
        ssc.awaitTermination()
    }


}
